package com.run.wz.test.hfile;

import java.io.IOException;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFile.Reader;
import org.apache.hadoop.hbase.io.hfile.HFileScanner;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.protobuf.Descriptors.FieldDescriptor;
import com.google.protobuf.GeneratedMessage;
import com.run.pbf.Bcp.Head;
import com.run.wz.center.utils.ProtocolType;

public class ReWrite {
	
	private static Configuration conf;
	private static Path path;
	private static HFileScanner scanner;
	private static HTable table;
	private static void init() throws IOException {
		conf = new Configuration();
		conf.set("hadoop.tmp.dir", "/data1");
		conf.set("fs.default.name", "hdfs://DG1HB1:9000");
		path = new Path("/hbase/actionlog1_201305/00526339d2e311bee5b76f3fecb6c630/1/fae917c3671e4600884bacc3d7646f45");
		FileSystem fileSystem = path.getFileSystem(conf);
		Reader createReader = HFile.createReader(fileSystem,path,new CacheConfig(conf) );
		scanner = createReader.getScanner(false, false, false);
		
		conf.set("hbase.zookeeper.quorum",
				"DG1HB3,DG1HB4,DG1HB5");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		
		
		table = new HTable(conf, "chenxu_secondary_index_1");
	}
//	public static void parseMsg()
	public static void main(String[] args) throws IOException {
		try {
			
			init();
			if(scanner.seekTo())
				while (scanner.next()) {
					KeyValue keyValue = scanner.getKeyValue();
					System.out.println(keyValue.getRow());
					byte[] qualifier = keyValue.getQualifier();
					ProtocolType protocolTypeByNum = ProtocolType.getProtocolTypeByNum(qualifier[0]);
					GeneratedMessage parseFrom = protocolTypeByNum.parseFrom(keyValue.getValue());
					Map<FieldDescriptor, Object> allFields = parseFrom.getAllFields();
					for(Entry<FieldDescriptor, Object> entry : allFields.entrySet()) {
						
						FieldDescriptor key = entry.getKey();
						Object value = entry.getValue();
						String string = value.toString();
						if(value instanceof Head) {
							Head head = (Head)value;
							Map<FieldDescriptor, Object> allFields2 = head.getAllFields();
							for(Entry<FieldDescriptor, Object> entry2 : allFields2.entrySet()) {
								FieldDescriptor key2 = entry2.getKey();
								Object value2 = entry2.getValue();
								Put put = new Put(Bytes.add(keyValue.getRow(), Bytes.toBytes(key2.getNumber()+"")));
								put.add(Bytes.toBytes("r"), Bytes.toBytes(value2.toString()), null);
								table.put(put);
							}
						} else {
							Put put = new Put(Bytes.add(keyValue.getRow(), Bytes.toBytes(key.getNumber()+"")));
							put.add(Bytes.toBytes("r"), Bytes.toBytes(string), null);
							table.put(put);
						}
					}
				}
			table.flushCommits();
		} finally {
			table.close();
		}
		
	}

}
